package com.tbit.uqbike.client.service.ordertrace.impl;

import com.alibaba.fastjson.JSON;
import com.tbit.uqbike.client.constant.RedisConstant;
import com.tbit.uqbike.client.dto.EventPutDTO;
import com.tbit.uqbike.client.dto.EventReportDTO;
import com.tbit.uqbike.client.dto.OrderTraceIdApplyDTO;
import com.tbit.uqbike.client.exception.BaseException;
import com.tbit.uqbike.client.ordertrace.Event;
import com.tbit.uqbike.client.ordertrace.EventHandleResult;
import com.tbit.uqbike.client.ordertrace.OrderStateMachine;
import com.tbit.uqbike.client.repository.OrderEventRepository;
import com.tbit.uqbike.client.service.event.EventHandleService;
import com.tbit.uqbike.client.service.ordertrace.OrderEventBO;
import com.tbit.uqbike.client.service.ordertrace.OrderTraceService;
import com.tbit.uqbike.client.service.ordertrace.PriorityQueueService;
import com.tbit.uqbike.client.service.ordertrace.business.AccountInfo;
import com.tbit.uqbike.client.service.ordertrace.business.ExtendedInfo;
import com.tbit.uqbike.client.service.ordertrace.business.OrderInfo;
import com.tbit.uqbike.client.service.ordertrace.business.UserInfo;
import com.tbit.uqbike.client.util.RedisLockUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 订单跟踪服务
 *
 *
 * @author <a href="mailto:chen.chaoxiong@tbit.com.cn">ChenChaoXiong</a>
 * @since 2024/7/2
 */
@Service
@Validated
@Slf4j
public class OrderTraceServiceImpl implements OrderTraceService {

    @Resource
    private OrderEventRepository orderEventRepository;

    @Autowired
    private RedisLockUtils redisLockUtils;

    @Autowired
    private PriorityQueueService priorityQueueService;

    @Autowired
    private EventHandleService eventHandleService;

    /**
     * 用于事务代理
     */
    @Lazy
    @Autowired
    private OrderTraceService self;

    @Resource
    private ThreadPoolExecutor orderTraceExecutor;

    @Value("${orderTrace.group.expireSecondTime:60}")
    private int expireSecondTime;

    @Value("${orderTrace.lock.retry.times:3}")
    private int lockRetryTimes;

    @Value("${orderTrace.lock.retry.interval:500}")
    private long lockRetryInterval;

    private static final String ORDER_TRACE_QUEUE_PREFIX = "orderTrace:priorityQueue:";

    private static final String ORDER_TRACE_RETRY_QUEUE_PREFIX = "order_trace_retry:";

    @Override
    public String applyOrderTraceId(OrderTraceIdApplyDTO apply) {
        String traceId = generateTraceId();
        OrderStateMachine orderStateMachine = orderEventRepository.createOrderStateMachine(traceId);


        AccountInfo accountInfo = AccountInfo.builder()
                                             .brandId(apply.getBrandId())
                                             .accountId(apply.getAccountId())
                                             .build();

        OrderEventBO orderEventBO = OrderEventBO.builder()
                .traceId(traceId)
                .orderStateMachine(orderStateMachine)
                .accountInfo(accountInfo)
                .extendedInfoMap(null)
                .build();


        orderEventRepository.saveOrUpdateOrderEvent(orderEventBO);

        return traceId;
    }

    @Override
    public void reportEvent(EventReportDTO eventReport) {
        log.info("eventReport:{}", eventReport);
        try {
            enqueueNormal(eventReport.getTraceId(), JSON.toJSONString(eventReport));
            asyncConsume(eventReport.getTraceId());
        } catch (Exception e) {
            log.error("reportEvent error", e);
            throw new BaseException(eventReport.getTraceId() + " reportEvent error");
        }
    }


    private void asyncConsume(String traceId) {
        orderTraceExecutor.execute(() -> {
            try {
                consume(traceId);
            } catch (Exception e) {
                log.error("Failed to consume event for traceId: {}", traceId, e);
            }
        });
    }


    /**
     * 重试三次获取锁，如果获取不到锁，则不再处理
     * 极端情况下，消息会丢失
     * @param traceId
     */
    private void consume(String traceId) {
        String lockValue = UUID.randomUUID().toString().replace("-", "");
        int retryCount = 0;
        
        while (retryCount < lockRetryTimes) {
            boolean lockStatus = redisLockUtils.lock(RedisConstant.ORDER_TRACE + traceId, lockValue);
            if (lockStatus) {
                try {
                    processEventQueueWithRetry(traceId);
                    return;
                } finally {
                    redisLockUtils.unLock(RedisConstant.ORDER_TRACE + traceId, lockValue);
                }
            }
            
            retryCount++;
            log.info("traceId:{} 获取锁失败，第{}次重试", traceId, retryCount);
            
            try {
                Thread.sleep(lockRetryInterval);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("Lock retry interrupted for traceId: {}", traceId, e);
            }
        }
        
        // 重试次数用完后，将不再尝试，极端情况下，消息会丢失
        log.warn("traceId:{} 获取锁失败超过最大重试次数，消息不再处理", traceId);
        
    }


    /**
     * 当消费失败时，将消费失败的事件放到重试队列中，等待下一次消费成功将重试队列中的所有事件重新入队
     * 避免因为同一个消息重复上报 或者 事件时间戳(作为score)相同造成乱序消费 最终导致流程受阻
     * @param traceId
     */
    
    private void processEventQueueWithRetry(String traceId) {
        while (true) {
            String eventReport = dequeueNormal(traceId);
            if (StringUtils.isBlank(eventReport)) {
                break;  // 队列为空时退出循环
            }
            
            log.info("consume traceId:{}  eventReport:{}", traceId, eventReport);
            
            EventReportDTO eventReportDTO = JSON.parseObject(eventReport, EventReportDTO.class);
            if (!accept(eventReportDTO)) {
                log.info("traceId:{} do not accept {}", traceId, eventReportDTO);
                enqueueRetry(traceId, eventReport);
            } else {
                try {
                    if (self.handleWithTransaction(eventReportDTO)) {
                        reEnqueueRetryToNormal(traceId);
                    }
                } catch (Exception e) {
                    log.error("Failed to handle event for traceId: {}", traceId, e);
                    // enqueueNormal(traceId, eventReport);
                }
            }
        }
    }

    /**
     * 保证handle方法在事务中执行
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public boolean handleWithTransaction(EventReportDTO eventReportDTO) {
        return handle(eventReportDTO);
    }


    /**
     * 将消费失败的事件放入重试队列
     * @param traceId
     * @param eventReport
     */
    private void enqueueRetry(String traceId, String eventReport){
        EventReportDTO eventReportDTO = JSON.parseObject(eventReport, EventReportDTO.class);
        String retryQueueKey = ORDER_TRACE_RETRY_QUEUE_PREFIX + traceId;
        priorityQueueService.enqueue(retryQueueKey, eventReport, eventReportDTO.getEventTimeMilli(),expireSecondTime);
    }

    /**
     * 从普通队列中获取事件
     * @param traceId
     * @return
     */
    private String dequeueNormal(String traceId){
        String normalQueueKey = ORDER_TRACE_QUEUE_PREFIX + traceId;
        return priorityQueueService.dequeue(normalQueueKey);
    }

    /**
     * 将事件放入普通队列
     * @param traceId
     * @param eventReport
     */
    private void enqueueNormal(String traceId, String eventReport){
        EventReportDTO eventReportDTO = JSON.parseObject(eventReport, EventReportDTO.class);
        String normalQueueKey = ORDER_TRACE_QUEUE_PREFIX + traceId;
        priorityQueueService.enqueue(normalQueueKey, eventReport, eventReportDTO.getEventTimeMilli(),expireSecondTime);
    }

    /**
     * 将重试队列中的所有事件重新放入普通队列
     * @param traceId
     */
    private void reEnqueueRetryToNormal(String traceId){
        String retryQueueKey = ORDER_TRACE_RETRY_QUEUE_PREFIX + traceId;
        Map<String, Double> messages = priorityQueueService.getAllMessages(retryQueueKey);
        for (Map.Entry<String, Double> entry : messages.entrySet()) {
            enqueueNormal(traceId, entry.getKey());
        }
    }



    

    /**
     * 判断是否接受事件
     *
     * todo 判断待校验事件与当前状态的相对关系，是否接受，如果不接受，判断待校验事件是否需要重试
     * 在当前状态之前的事件丢弃，当前状态之后的事件重试
     * @param eventReport
     * @return
     */
    private boolean accept(EventReportDTO eventReport) {
        try {
            Event event = new Event(eventReport.getEventType(),eventReport.getBusinessInfo());
            OrderEventBO orderEventBO = orderEventRepository.queryOrderEventByTraceId(eventReport.getTraceId());
            OrderStateMachine orderStateMachine = orderEventBO.getOrderStateMachine();
            if (!orderStateMachine.accept(event)){
                log.info("OrderStateMachine {} do not accept event {}",orderStateMachine,event);
                return false;
            }
        }catch (Exception e){
            log.error("traceId {} handleEvent {} error",eventReport.getTraceId(),JSON.toJSONString(eventReport), e);
            return false;
        }
        return true;
    }

    private boolean handle(EventReportDTO eventReportDTO){

            Event event = new Event(eventReportDTO.getEventType(),eventReportDTO.getBusinessInfo());
            OrderEventBO orderEventBO = orderEventRepository.queryOrderEventByTraceId(eventReportDTO.getTraceId());
            OrderStateMachine orderStateMachine = orderEventBO.getOrderStateMachine();
            String requestId = UUID.randomUUID().toString().replace("-", "");
            EventHandleResult handleResult = orderStateMachine.handleEvent(event,requestId);
            ExtendedInfo extendedInfo = ExtendedInfo.builder()
                                                    .eventTime(eventReportDTO.getEventTime())
                                                    .detail(eventReportDTO.getDetail())
                                                    .message(eventReportDTO.getMessage())
                                                    .businessInfo(eventReportDTO.getBusinessInfo())
                                                    .eventDesc(eventReportDTO.getEventDesc())
                                                    .build();
            Map<String, ExtendedInfo> extendedInfoMap = new HashMap<>();
            extendedInfoMap.put(requestId, extendedInfo);
            orderEventBO.setExtendedInfoMap(extendedInfoMap);
            UserInfo userInfo = UserInfo.builder()
                                        .userId(eventReportDTO.getUserId())
                                        .userPhone(eventReportDTO.getUserPhone())
                                        .build();
            OrderInfo orderInfo = OrderInfo.builder()
                                           .machineId(eventReportDTO.getMachineId())
                                           .orderNo(eventReportDTO.getOrderNo())
                                           .build();
            orderEventBO.setOrderInfo(orderInfo);
            orderEventBO.setUserInfo(userInfo);
            if (handleResult.isAccept()) {
                orderEventRepository.saveOrUpdateOrderEvent(orderEventBO);
            } else {
                throw new BaseException(handleResult.getMsg());
            }
        return true;
    }

    private String generateTraceId() {
        return UUID.randomUUID().toString().replace("-", "");
    }


    @Override
    public void putEvent(EventPutDTO eventPutDTO) {
        eventHandleService.handle(eventPutDTO);
    }
}
